package com.ilearn.zookeeper.distributionlock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * @Project: seb-core
 * @Description:
 * @Version 1.0
 * @Throws SystemException:
 * @Author: <li>2018/12/25 lyndon Create 1.0
 * @Copyright (c)2018-2018 中央结算公司，版权所有
 * @Modified By:
 */
public class DistributedLock implements Lock, Watcher {

    private ZooKeeper zk = null;
    private String ROOT_LOCK="/locks"; //定义根节点
    private String WAIT_LOCK;  //等待前一个节点锁
    private String CURRENT_LOCK; //标识当前的锁

    private CountDownLatch countDownLatch;

    public DistributedLock() {
        try {
            // 先创建连接
            zk = new ZooKeeper("192.168.11.153:2181",
                    4000, this);
            // 判断根节点是否存在, 不存在就创建根节点
                Stat stat = zk.exists(ROOT_LOCK, false);
                if (stat==null){
                    zk.create(ROOT_LOCK, "0".getBytes(),
                            ZooDefs.Ids.READ_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void lock() {
        // 如果获得锁成功
        if (tryLock()){
            System.out.println(Thread.currentThread().getName() + " -> " + CURRENT_LOCK + " -> 获得锁成功");
            return;
        }
        // 如果没有获得锁，就一直等待
        try {
            waitForLock(WAIT_LOCK);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 这里就是阻塞等待的过程
     * @param prev
     * @return
     * @throws KeeperException
     * @throws InterruptedException
     */
    private boolean waitForLock(String prev) throws KeeperException, InterruptedException {
        Stat stat = zk.exists(prev, true); // 监听当前节点的上一个节点
        if (stat != null){
            System.out.println(Thread.currentThread().getName() + " -> 等待锁" + ROOT_LOCK + "/" + prev + "释放");
            // 等待的过程就要用到 CountDownLatch
            countDownLatch = new CountDownLatch(1);
            countDownLatch.await();
            System.out.println(Thread.currentThread().getName() + "-> 获得锁成功");
        }
        return true;
    }


    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        // 创建临时有序节点
        try {
            CURRENT_LOCK = zk.create(ROOT_LOCK+"/", "0".getBytes(),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(Thread.currentThread().getName() + " -> " +
                    CURRENT_LOCK + "创建竞争锁");
            // 获取根节点下的所有子节点
            List<String> childrens = zk.getChildren(ROOT_LOCK, false);
            // 定义一个集合进行排序
            SortedSet<String> sortedSet = new TreeSet<String>();
            for (String children : childrens) {
                sortedSet.add(ROOT_LOCK + "/" + children);
            }
            final String firstNode = sortedSet.first();// 获得当前所有子节点中最小的节点
            // 返回当前节点的前一个节点
            SortedSet<String> lessThenMe = ((TreeSet<String>) sortedSet).headSet(CURRENT_LOCK);
            // 通过当前的节点和子节点中最小的节点进行比较，
            // 如果相等，表示获得锁成功
            if (CURRENT_LOCK.equals(firstNode)){
                return true;
            }
            // 获得比当前节点更小的最后一个节点，设置给WAIT_LOCK
            if (! lessThenMe.isEmpty()){
                WAIT_LOCK=lessThenMe.last();
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        System.out.println(Thread.currentThread().getName() + " -> 释放锁" + CURRENT_LOCK);
        try {
            zk.delete(CURRENT_LOCK, -1);
            CURRENT_LOCK=null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    /**
     * 处理 waitForLock当中，监听等待上一个节点锁的事件
     * @param watchedEvent
     */
    @Override
    public void process(WatchedEvent watchedEvent) {
        if (this.countDownLatch != null){
            this.countDownLatch.countDown();
        }
    }
}
